package com.dvj.demo.rabbitmq;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;


/**
 * RabbitMQListener
 *
 * @author Dvj
 * @date 2021/7/1 18:41
 */
@Component
public class RabbitMQListener {

    @RabbitListener(queues = "boot_queue")
    public void listenerQueue(Message message) {
        System.out.println(new String(message.getBody()));
    }

    /**
     * Consumer ACK机制
     * 1.设置手动签收：ackMode="manual"
     * 2.让监听器类实现ChannelAwareMessageListener接口
     * 3.如果消息处理成功，则调用channel的basicAck()签收
     * 4.如果消息处理失败，则调用channel的basicNack()拒绝签收，broker重新发送给consumer
     *
     * @param message
     */
    @RabbitListener(queues = "boot_queue_confirm", ackMode = "MANUAL")
    public void ackListener(Message message, Channel channel) throws IOException, InterruptedException {
        Thread.sleep(1000);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 1.接收转换消息
        System.out.println(new String(message.getBody()));

        // 2.处理业务逻辑
        System.out.println("处理业务逻辑...");

        // 3.手动签收
        try {
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            // 4.拒绝签收
            /*
            basicNack(long deliveryTag, boolean multiple, boolean requeue)
            参数：
                requeue：重回队列，如果为true，则消息重新回到queue
             */
            channel.basicNack(deliveryTag, true, true);
        }
    }

    /**
     * Consumer 限流机制
     * 1.确保ack机制为手动确认
     * 2.listener-container配置属性，prefetch = 1 表示消费端每次从mq拉取一条消息来消费，
     * 直到手动确认消费完毕后才会继续拉去下一条消息。
     *
     * @param data
     * @param message
     * @param channel
     */
    @RabbitListener(queues = "boot_queue_qos", ackMode = "MANUAL", containerFactory = "qosContainerFactory")
    public void qosListener(String data, Message message, Channel channel) throws IOException, InterruptedException {
        // 1.获取消息
        System.out.println(data);
        // 2.处理业务逻辑
        System.out.println("业务处理");

        Thread.sleep(1000);

        // 3.签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
    }

    @RabbitListener(queues = "boot_queue_ttl", ackMode = "MANUAL")
    public void ttlListener(String data, Message message, Channel channel) throws IOException, InterruptedException {
        Thread.sleep(1000);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 1.接收转换消息
        System.out.println(new String(message.getBody()));

        // 2.处理业务逻辑
        System.out.println("处理业务逻辑...");

        // 3.手动签收
        try {
            int i = 3 / 0;
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            // 4.拒绝签收
            System.out.println("出现异常，拒绝接受");
            /*
            basicNack(long deliveryTag, boolean multiple, boolean requeue)
            参数：
                requeue：重回队列，如果为true，则消息重新回到queue
             */
            channel.basicNack(deliveryTag, true, false);
        }
    }
}
